package cn.dawn.webcollector.fetcher;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.dawn.webcollector.util.BerkeleyDBUtils;
import cn.dawn.webcollector.util.Config;

import com.sleepycat.je.Cursor;
import com.sleepycat.je.Database;
import com.sleepycat.je.DatabaseEntry;
import com.sleepycat.je.Environment;
import com.sleepycat.je.LockMode;
import com.sleepycat.je.OperationStatus;

/**
 * 
 * <p>任务更新器</p>
 * <p>任务更新器，用来更新任务的状态和加入新的任务，网页爬取成功后需要更新CrawlDatum中的状态，对网页做解析，发现新的连接，也需要更新CrawlDatum。</p>
 * @Copyright:Copyright (c) 2015
 * @Author:wangxiao
 * @Version:1.0
 * @Time:2015年10月10日
 */
public class DbUpdater {

	public static final Logger LOG = LoggerFactory.getLogger(DbUpdater.class);

	Environment env;
	SegmentWriter segmentWriter;
	Database lockDatabase;

	public DbUpdater(Environment env) {
		this.env = env;
		segmentWriter = new SegmentWriter(env);
	}

	public void lock() throws Exception {
		lockDatabase = env.openDatabase(null, Config.BDB_LOCK, BerkeleyDBUtils.defaultDBConfig);
		DatabaseEntry key = new DatabaseEntry(Config.BDB_LOCK.getBytes("utf-8"));
		DatabaseEntry value = new DatabaseEntry(Config.BDB_LOCKED.getBytes("utf-8"));
		lockDatabase.put(null, key, value);
		lockDatabase.sync();
		lockDatabase.close();
	}

	public boolean isLocked() throws Exception {
		boolean isLocked = false;
		lockDatabase = env.openDatabase(null, Config.BDB_LOCK, BerkeleyDBUtils.defaultDBConfig);
		DatabaseEntry key = new DatabaseEntry(Config.BDB_LOCK.getBytes("utf-8"));
		DatabaseEntry value = new DatabaseEntry();
		if (lockDatabase.get(null, key, value, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
			String lockInfo = new String(value.getData(), "utf-8");
			if (lockInfo.equals(Config.BDB_LOCKED)) {
				isLocked = true;
			}
		}
		lockDatabase.close();
		return isLocked;
	}

	public void unlock() throws Exception {

		lockDatabase = env.openDatabase(null, Config.BDB_LOCK, BerkeleyDBUtils.defaultDBConfig);
		DatabaseEntry key = new DatabaseEntry(Config.BDB_LOCK.getBytes("utf-8"));
		DatabaseEntry value = new DatabaseEntry(Config.BDB_UNLOCK.getBytes("utf-8"));
		lockDatabase.put(null, key, value);
		lockDatabase.sync();
		lockDatabase.close();

	}

	public void close() throws Exception {
		segmentWriter.close();
	}

	public void merge() throws Exception {
		LOG.info("start merge");
		Database crawldbDatabase = env.openDatabase(null, Config.BDB_CRAWLDB, BerkeleyDBUtils.defaultDBConfig);
		/* 合并fetch库 */
		LOG.info("merge fetch database");
		Database fetchDatabase = env.openDatabase(null, Config.BDB_FETCH, BerkeleyDBUtils.defaultDBConfig);
		Cursor fetchCursor = fetchDatabase.openCursor(null, null);
		DatabaseEntry key = new DatabaseEntry();
		DatabaseEntry value = new DatabaseEntry();
		while (fetchCursor.getNext(key, value, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
			crawldbDatabase.put(null, key, value);
		}
		fetchCursor.close();
		fetchDatabase.close();
		/* 合并link库 */
		LOG.info("merge link database");
		Database linkDatabase = env.openDatabase(null, Config.BDB_LINK, BerkeleyDBUtils.defaultDBConfig);
		Cursor linkCursor = linkDatabase.openCursor(null, null);
		while (linkCursor.getNext(key, value, LockMode.DEFAULT) == OperationStatus.SUCCESS) {
			if (!(crawldbDatabase.get(null, key, value, LockMode.DEFAULT) == OperationStatus.SUCCESS)) {
				crawldbDatabase.put(null, key, value);
			}
		}
		linkCursor.close();
		linkDatabase.close();
		LOG.info("end merge");
		crawldbDatabase.sync();
		crawldbDatabase.close();

		env.removeDatabase(null, Config.BDB_FETCH);
		LOG.debug("remove fetch database");
		env.removeDatabase(null, Config.BDB_LINK);
		LOG.debug("remove link database");

	}

	public SegmentWriter getSegmentWriter() {
		return segmentWriter;
	}

}
